import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author wangzj
 * @description
 * @date 23:57 2020/7/6
 */
public class StreamWordCount {
    public static void main(String[] args) throws Exception {
        //创建流式计算环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置时间属性为 EventTime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //监控本地的9000端口
        DataStream text = env.socketTextStream("127.0.0.1", 9000, "\n");
        //将接收的数据进行拆分，分组，窗口计算并且进行聚合输出
        DataStream<WordCount> windowsCount = text
                .flatMap(new FlatMapFunction<String, WordCount>() {
                    public void flatMap(String value, Collector<WordCount> out) {
                        for (String word : value.split("\\s")) {
                            out.collect(new WordCount(word, 1L));
                        }
                    }
                })
                .keyBy("word")
                //滑动窗口，每一秒统计一次最近5s的数据（总共会统计5次）
                //窗口时间为5s，滑动时间为1s
                .timeWindow(Time.seconds(5), Time.seconds(2))
                .reduce(new ReduceFunction<WordCount>() {
                    @Override
                    public WordCount reduce(WordCount value1, WordCount value2) {
                        return new WordCount(value1.word, value1.count + value2.count);
                    }
                });
        //打印结果
        windowsCount.print().setParallelism(1);
        env.execute("Stream WordCount");
    }

    /**
     * 求和
     */
    public static class WordCount {
        public String word;
        public long count;

        public WordCount() {

        }

        public WordCount(String word, long count) {
            this.word = word;
            this.count = count;
        }

        @Override
        public String toString() {
            return word + ":" + count;
        }
    }
}
